/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * ---------------------------------------------------------------------------------------
 *
 * imcs_insert.cpp
 * routines to support InMemoryColStore
 *
 * IDENTIFICATION
 * src/gausskernel/storage/imcstore/imcs_insert.cpp
 *
 * ---------------------------------------------------------------------------------------
 */

#include "postgres.h"
#include "knl/knl_variable.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "access/genam.h"
#include "access/cstore_delta.h"
#include "access/reloptions.h"
#include "catalog/catalog.h"
#include "utils/aiomem.h"
#include "utils/datum.h"
#include "utils/gs_bitmap.h"
#include "utils/fmgroids.h"
#include "utils/snapmgr.h"
#include "utils/relcache.h"
#include "catalog/pg_type.h"
#include "access/imcs/imcs_am.h"
#include "access/imcs/imcu_storage.h"
#include "utils/builtins.h"
#include "executor/executor.h"
#include "replication/dataqueue.h"
#include "storage/lmgr.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "pgxc/pgxc.h"
#include "access/heapam.h"
#include "utils/memutils.h"
#include "utils/date.h"
#include "access/imcs/imcs_allocspace.h"
#include "storage/ipc.h"
#include "catalog/pg_partition_fn.h"
#include "libpq/pqformat.h"
#include "workload/workload.h"
#include "commands/tablespace.h"
#include "optimizer/var.h"
#include "catalog/index.h"
#include "storage/time_series_compress.h"
#include "access/imcs/imcs_insert.h"
#include "access/imcs/imcs_am.h"
#include "access/imcs/imcs.h"
#include "utils/dynahash.h"

#define MAX_CU_WRITE_REQSIZ 64

#define cstore_backwrite_quantity 8192

#define ENABLE_DELTA(batch)                                                              \
    ((batch) != NULL && ((g_instance.attr.attr_storage.enable_delta_store && IsEnd()) && \
        ((batch)->m_rows_curnum < m_delta_rows_threshold)))

/* total memory cache size by adio, used for memory control with  cstore_backwrite_max_threshold */
int64 hstore_adio_write_cache_size = 0;

extern void CUListWriteAbort(int code, Datum arg);

static inline void CuAppendNullValue(int which, void *cu_ptr)
{
    ((IMCU *)cu_ptr)->AppendNullValue(which);
}

static inline int str_to_uint64(const char *str, uint64 *val)
{
    char *end = NULL;
    uint64 value = 0;

    Assert(str != NULL && val != NULL);
    uint32 str_len = strlen(str);
    if (str_len == 0) {
        return -1;
    }

    /* clear errno before convert */
    errno = 0;
#ifdef WIN32
    value = _strtoui64(str, &end, 10);
#else
    value = strtoul(str, &end, 10);
#endif
    if ((errno != 0) || (end != (str + str_len))) {
        return -1;
    }

    *val = value;
    return 0;
}

/* index of m_formCUFuncArray[] calls */
#define FORMCU_IDX_NONE_NULL 0
#define FORMCU_IDX_HAVE_NULL 1

/*
 * @Description: compute the max number of keys within all index relation.
 * @IN rel: result relation info
 * @Return: the max number of index keys
 * @See also:
 */
static inline int get_max_num_of_index_keys(ResultRelInfo *rel)
{
    IndexInfo **indexes = rel->ri_IndexRelationInfo;
    int max_num = indexes[0]->ii_NumIndexAttrs;

    for (int i = 1; i < rel->ri_NumIndices; ++i) {
        if (max_num < indexes[i]->ii_NumIndexAttrs) {
            /* update the max number */
            max_num = indexes[i]->ii_NumIndexAttrs;
        }
    }
    return max_num;
}

/*
 * @Description: check whether result relation has any index.
 * @IN rel: result relation info
 * @Return: true if have any index; otherwise false.
 * @See also:
 */
static inline bool relation_has_indexes(ResultRelInfo *rel)
{
    return (rel && rel->ri_NumIndices > 0);
}


IMCSInsert::IMCSInsert(_in_ Relation relation, _in_ ResultRelInfo *result_rel_info, _in_ bool is_update_cu,
    _in_ Plan *plan, _in_ ImcsMemInfoArg *argmem_info, TupleDesc tupdesc_with_ctid)
    : m_fullCUSize(RelationGetMaxBatchRows(relation)), m_delta_rows_threshold(RelationGetDeltaRowsThreshold(relation))
{
    m_slot_id = -1;
    m_insert_end_flag = false;
    m_relation = relation;
    td_tid = tupdesc_with_ctid;
    m_resultRelInfo = result_rel_info;
    m_bufferedBatchRows = NULL;
    int max_values_count = RelationGetMaxBatchRows(relation);
    if (tupledesc_have_pck(relation->rd_att->constr)) {
        m_tmpBatchRows = New(CurrentMemoryContext)IMCSBulkloadRows(relation->rd_att, max_values_count, true);
    }

    /* create BatchRows for index, if relation has index. */
    if (relation_has_indexes(result_rel_info)) {
        /* find max number of index keys,
         * and we will reuse these batch buffers.
         */
        int max_key_num = get_max_num_of_index_keys(result_rel_info);
        m_idxBatchRow = imcs_bulkload_indexbatch_init(max_key_num, max_values_count);
    }
    // m_tmpBatchRows = args.tmp_batch_rows;
    // m_idxBatchRow = args.idx_batch_row;
    m_cstorInsertMem = NULL;
    m_cur_cuid = 0;
    SpinLockInit(&populate_mutex);

    /* set the cstore Insert mem info. */
    InitInsertMemArg(plan, argmem_info);

    m_tmpMemCnxt = AllocSetContextCreate(CurrentMemoryContext, "INSERT TEMP MEM CNXT", ALLOCSET_DEFAULT_MINSIZE,
        ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);
    m_batchInsertCnxt = AllocSetContextCreate(CurrentMemoryContext, "Batch Insert Mem CNXT", ALLOCSET_DEFAULT_MINSIZE,
        ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, STANDARD_CONTEXT, m_cstorInsertMem->MemInsert * 1024L);

    ADIO_RUN()
    {
        m_aio_memcnxt = AllocSetContextCreate(CurrentMemoryContext, "ADIO IMCU CACHE CNXT", ALLOCSET_DEFAULT_MINSIZE,
            ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);
        /* the other ADIO vars will be initialized later */
    }
    ADIO_ELSE()
    {
        m_aio_memcnxt = NULL;
        m_aio_cu_pptr = NULL;
        m_aio_dispath_cudesc = NULL;
        m_aio_dispath_idx = NULL;
        m_aio_cache_write_threshold = NULL;
        m_vfdList = NULL;
    }
    ADIO_END();

    AutoContextSwitch auto_mem_context(m_batchInsertCnxt);

    int attNo = td_tid->natts;

    m_cuStorage = (IMCUStorage **)palloc(sizeof(IMCUStorage *) * attNo);
    m_cuCmprsOptions = (compression_options *)palloc(sizeof(compression_options) * attNo);

    for (int i = 0; i < attNo; ++i) {
        if (td_tid->attrs[i].attisdropped) {
            m_cuStorage[i] = NULL;
        } else {
            /* Here we must use physical column id */
            CFileNode cfile_node(m_relation->rd_node, td_tid->attrs[i].attnum, MAIN_FORKNUM);
            m_cuStorage[i] = New(CurrentMemoryContext)IMCUStorage(cfile_node);
        }
        /* init compression filter */
        m_cuCmprsOptions[i].reset();
    }

    /* set the compression level and extra modes. */
    m_compress_modes = 0;
    heaprel_set_compressing_modes(m_relation, &m_compress_modes);

    /* set update flag */
    m_isUpdate = is_update_cu;

    BeginBatchInsert();
}

IMCSInsert::~IMCSInsert()
{
    m_relation = NULL;
    m_idxBatchRow = NULL;
    m_idxRelation = NULL;
    m_setMinMaxFuncs = NULL;
    m_fake_isnull = NULL;
    m_idxInsertArgs = NULL;
    m_aio_memcnxt = NULL;
    m_fake_values = NULL;
    m_delta_relation = NULL;
    m_cuCmprsOptions = NULL;
    m_estate = NULL;
    m_cuDescPPtr = NULL;
    m_delta_desc = NULL;
    m_aio_dispath_idx = NULL;
    m_tmpBatchRows = NULL;
    m_bufferedBatchRows = NULL;
    m_batchInsertCnxt = NULL;
    m_cuStorage = NULL;
    m_idxKeyAttr = NULL;
    m_resultRelInfo = NULL;
    m_tmpMemCnxt = NULL;
    m_aio_dispath_cudesc = NULL;
    m_vfdList = NULL;
    m_cuPPtr = NULL;
    m_idxKeyNum = NULL;
    m_aio_cache_write_threshold = NULL;
    m_formCUFuncArray = NULL;
    m_econtext = NULL;
    m_aio_cu_pptr = NULL;
    m_cstorInsertMem = NULL;
    m_idxInsert = NULL;
}

void IMCSInsert::FreeMemAllocateByAdio()
{
    int attNo = td_tid->natts;

    if (m_aio_cu_pptr) {
        for (int i = 0; i < attNo; i++) {
            if (m_aio_cu_pptr[i]) {
                pfree(m_aio_cu_pptr[i]);
                m_aio_cu_pptr[i] = NULL;
            }
        }
        pfree(m_aio_cu_pptr);
        m_aio_cu_pptr = NULL;
    }
    if (m_aio_dispath_cudesc) {
        for (int i = 0; i < attNo; i++) {
            if (m_aio_dispath_cudesc[i]) {
                pfree(m_aio_dispath_cudesc[i]);
                m_aio_dispath_cudesc[i] = NULL;
            }
        }
        pfree(m_aio_dispath_cudesc);
        m_aio_dispath_cudesc = NULL;
    }
    if (m_aio_dispath_idx) {
        pfree(m_aio_dispath_idx);
        m_aio_dispath_idx = NULL;
    }
    if (m_aio_cache_write_threshold) {
        pfree(m_aio_cache_write_threshold);
        m_aio_cache_write_threshold = NULL;
    }

    if (m_vfdList) {
        for (int i = 0; i < attNo; i++) {
            if (m_vfdList[i]) {
                pfree(m_vfdList[i]);
                m_vfdList[i] = NULL;
            }
        }
        pfree(m_vfdList);
        m_vfdList = NULL;
    }

    return;
}

/*
 * @Description: set m_formCUFuncArray[] function point.
 * @See also:
 */
inline void IMCSInsert::SetFormCUFuncArray(Form_pg_attribute attr, int col)
{
    if (ATT_IS_NUMERIC_TYPE(attr->atttypid) && attr->atttypmod != -1) {
        /* Numeric data type. */
        m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_NONE_NULL] = &IMCSInsert::FormCUTNumeric<false>;
        m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_HAVE_NULL] = &IMCSInsert::FormCUTNumeric<true>;
    } else if (ATT_IS_CHAR_TYPE(attr->atttypid)) {
        /* String data type. */
        m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_NONE_NULL] = &IMCSInsert::FormCUTNumString<false>;
        m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_HAVE_NULL] = &IMCSInsert::FormCUTNumString<true>;
    } else {
        /* copy data directly for the other data types */
        m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_NONE_NULL] = &IMCSInsert::FormCUT<false>;
        m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_HAVE_NULL] = &IMCSInsert::FormCUT<true>;
    }
}

void IMCSInsert::Destroy()
{
    /* Step 1: End of batch insert */
    EndBatchInsert();

    /* Step 2: Destroy all the new objects. */
    int attNo = td_tid->natts;
    for (int col = 0; col < attNo; ++col) {
        if (m_cuDescPPtr[col]) {
            delete m_cuDescPPtr[col];
            m_cuDescPPtr[col] = NULL;
        }
        if (m_cuStorage[col])
            DELETE_EX(m_cuStorage[col]);
    }

    if (relation_has_indexes(m_resultRelInfo)) {
        for (int i = 0; i < m_resultRelInfo->ri_NumIndices; ++i) {
            /*
             * For partition table with cbtree/cgin index, release the fake dummy relation,
             * otherwise just close the index relation .
             */
            if ((m_idxRelation[i]->rd_rel->relam == CBTREE_AM_OID || m_idxRelation[i]->rd_rel->relam == CGIN_AM_OID) &&
                RelationIsPartition(m_relation)) {
                releaseDummyRelation(&m_idxRelation[i]);
            } else {
                relation_close(m_idxRelation[i], RowExclusiveLock);
            }

            if (m_deltaIdxRelation[i]) {
                relation_close(m_deltaIdxRelation[i], RowExclusiveLock);
            }

            /* destroy index inserter */
            if (m_idxInsert[i] != NULL) {
                DELETE_EX(m_idxInsert[i]);
                /* destroy index inserter' arguments */
                DeInitInsertArg(m_idxInsertArgs[i]);
            }
        }

        FreeExecutorState(m_estate);
        m_estate = NULL;
        m_econtext = NULL;
        pfree(m_fake_values);
        pfree(m_fake_isnull);
    }

#ifdef USE_ASSERT_CHECKING
    if (m_cuPPtr) {
        BFIO_RUN()
        {
            for (int i = 0; i < attNo; i++) {
                Assert(m_cuPPtr[i] == NULL);
            }
        }
        BFIO_END();
    }
#endif

    ADIO_RUN()
    {
        FreeMemAllocateByAdio();
        MemoryContextDelete(m_aio_memcnxt);
    }
    ADIO_END();

    /* Step 3: Destroy memory context */
    MemoryContextDelete(m_tmpMemCnxt);
    MemoryContextDelete(m_batchInsertCnxt);

    /* Step 4: Reset all the pointers */
    m_formCUFuncArray = NULL;
    m_setMinMaxFuncs = NULL;
    m_cuStorage = NULL;
    m_cuDescPPtr = NULL;
    m_cuPPtr = NULL;
    m_idxKeyAttr = NULL;
    m_idxKeyNum = NULL;
    m_idxRelation = NULL;
    m_cuCmprsOptions = NULL;
    m_fake_values = NULL;
    m_fake_isnull = NULL;

    if (m_cstorInsertMem) {
        pfree_ext(m_cstorInsertMem);
    }
}

/*
 * @Description: init insert memory info for cstore insert. There are three branches, plan is the optimizer
 * estimation parameter passed to the storage layer for execution; argmem_info is to execute the operator
 * 	 from the upper layer to pass the parameter to the insert; others is uncontrolled memory.
 * @IN plan: If insert operator is directly used, the plan mem_info is used.
 * @IN argmem_info: argmem_info is used to passing parameters, such as update.
 * @Return: void
 * @See also: InitInsertPartMemArg
 */
void IMCSInsert::InitInsertMemArg(Plan *plan, ImcsMemInfoArg *argmem_info)
{
    bool has_pck = false;
    int partial_cluster_rows = 0;

    /* get cluster key */
    partial_cluster_rows = RelationGetPartialClusterRows(m_relation);
    if (m_relation->rd_rel->relhasclusterkey) {
        has_pck = true;
    }

    m_cstorInsertMem = (ImcsMemInfoArg *)palloc0(sizeof(struct ImcsMemInfoArg));
    if (plan != NULL && plan->operatorMemKB[0] > 0) {
        if (!has_pck) {
            m_cstorInsertMem->MemInsert = plan->operatorMemKB[0];
            m_cstorInsertMem->MemSort = 0;
        } else {
            m_cstorInsertMem->MemInsert = (int)((double)plan->operatorMemKB[0] * (double)(m_fullCUSize * 3) /
                (double)(m_fullCUSize * 3 + partial_cluster_rows));
            m_cstorInsertMem->MemSort = plan->operatorMemKB[0] - m_cstorInsertMem->MemInsert;
        }
        m_cstorInsertMem->canSpreadmaxMem = plan->operatorMaxMem;
        m_cstorInsertMem->spreadNum = 0;
        m_cstorInsertMem->partitionNum = 1;
        MEMCTL_LOG(DEBUG2,
            "IMCSInsert(init plan):Insert workmem is : %dKB, sort workmem: %dKB,can spread "
            "mem is %dKB.",
            m_cstorInsertMem->MemInsert, m_cstorInsertMem->MemSort, m_cstorInsertMem->canSpreadmaxMem);
    } else if (argmem_info != NULL) {
        Assert(argmem_info->partitionNum > 0);
        m_cstorInsertMem->canSpreadmaxMem = argmem_info->canSpreadmaxMem;
        m_cstorInsertMem->MemInsert = argmem_info->MemInsert;
        m_cstorInsertMem->MemSort = argmem_info->MemSort / argmem_info->partitionNum;
        if (m_cstorInsertMem->MemSort < SORT_MIM_MEM)
            m_cstorInsertMem->MemSort = SORT_MIM_MEM;
        m_cstorInsertMem->spreadNum = argmem_info->spreadNum;
        m_cstorInsertMem->partitionNum = argmem_info->partitionNum;
        MEMCTL_LOG(DEBUG2,
            "IMCSInsert(init argmem_info):Insert workmem is : %dKB, one paritition sort workmem: %dKB,"
            "partition totalnum is %d, can spread mem is %dKB.",
            m_cstorInsertMem->MemInsert, m_cstorInsertMem->MemSort, m_cstorInsertMem->partitionNum,
            m_cstorInsertMem->canSpreadmaxMem);
    } else {
        m_cstorInsertMem->canSpreadmaxMem = 0;
        m_cstorInsertMem->MemInsert = u_sess->attr.attr_storage.partition_max_cache_size;
        m_cstorInsertMem->MemSort = u_sess->attr.attr_storage.psort_work_mem;
        m_cstorInsertMem->spreadNum = 0;
        m_cstorInsertMem->partitionNum = 1;
    }
}
void IMCSInsert::InitFuncPtr()
{
    // MemoryContext oldcontext = NULL;
    int attNo = td_tid->natts;
    FormData_pg_attribute *attrs = td_tid->attrs;

    m_setMinMaxFuncs = (ImcsFuncSetMinMax *)palloc(attNo * sizeof(ImcsFuncSetMinMax));
    m_formCUFuncArray = (FormCUFuncArray *)palloc(sizeof(FormCUFuncArray) * attNo);
    m_cuDescPPtr = (IMCUDesc **)palloc(attNo * sizeof(IMCUDesc *));

    /*
     * Initilize Min/Max set function for all columns
     * Initilize FormCU function
     */
    for (int col = 0; col < attNo; ++col) {
        if (!attrs[col].attisdropped) {
            m_setMinMaxFuncs[col] = ImcsGetMinMaxFunc(attrs[col].atttypid);
            SetFormCUFuncArray(&attrs[col], col);
            // oldcontext = MemoryContextSwitchTo(IMCU_CACHE->im_cudesc_hash->hcxt);
            m_cuDescPPtr[col] = New(CurrentMemoryContext)IMCUDesc;
            // (void)MemoryContextSwitchTo(oldcontext);
        } else {
            m_setMinMaxFuncs[col] = NULL;
            m_cuDescPPtr[col] = NULL;
            m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_NONE_NULL] = NULL;
            m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_HAVE_NULL] = NULL;
        }
    }
}

void IMCSInsert::InitColSpaceAlloc()
{
    int attNo = td_tid->natts;
    FormData_pg_attribute *attrs = td_tid->attrs;

    AttrNumber *attrIds = (AttrNumber *)palloc(sizeof(AttrNumber) * attNo);
    for (int i = 0; i < attNo; ++i) {
        attrIds[i] = attrs[i].attnum;

        /* Set all column use APPEND_ONLY */
        if (!attrs[i].attisdropped)
            m_cuStorage[i]->SetAllocateStrategy(APPEND_ONLY);
    }

    /* when we get the max cu ID, wo need to check the TIDS in all the btree index */
    List *index_rel = NIL;
    if (m_resultRelInfo != NULL) {
        for (int i = 0; i < m_resultRelInfo->ri_NumIndices; ++i) {
            Oid am_oid = m_idxRelation[i]->rd_rel->relam;
            if (am_oid == CBTREE_AM_OID || am_oid == CGIN_AM_OID) {
                index_rel = lappend(index_rel, m_idxRelation[i]);
            }
        }
    }

    /* Whether we need build column space cache for relation */

    pfree_ext(attrIds);
    if (index_rel != NIL) {
        list_free(index_rel);
    }
}

void IMCSInsert::BeginBatchInsert()
{
    /* Step 1: Init function pointer for FormCU and get min/max */
    InitFuncPtr();

    /*
     * Step 2: Init partial cluster key
     * If relation has cluster keys, then initialize env.
     */

    m_bufferedBatchRows = New(CurrentMemoryContext)IMCSBulkloadRows(td_tid, RelationGetMaxBatchRows(m_relation), true);

    /* Step3: Initilize delta information */
    InitDeltaInfo();

    /* Step 4: Initilize index information if needed */
    InitIndexInfo();

    /* Step 5: Initialize column space cache allocation */
    InitColSpaceAlloc();

    /* Step 6: Initilize IMCU objects. */
    m_cuPPtr = (IMCU **)palloc0(sizeof(IMCU *) * td_tid->natts);

    /*
     * Step 7: Lock relfilenode.
     * When one column data insert a cu block, catchup read it immediately,
     * catchup maybe get zero block. This case is tested at xfs file system.
     * So insert acquire relfilenode lock to block catchup.
     */
    LockRelFileNode(m_relation->rd_node, RowExclusiveLock);

    ADIO_RUN()
    {
        m_aio_dispath_idx = (int *)palloc0(sizeof(int) * td_tid->natts);
        m_aio_dispath_cudesc = (AioDispatchCUDesc_t ***)palloc(sizeof(AioDispatchCUDesc_t **) * td_tid->natts);
        m_aio_cu_pptr = (IMCU ***)palloc(sizeof(IMCU **) * td_tid->natts);
        m_aio_cache_write_threshold = (int32 *)palloc0(sizeof(int32) * td_tid->natts);

        m_vfdList = (File **)palloc(sizeof(File *) * td_tid->natts);

        for (int col = 0; col < td_tid->natts; ++col) {
            m_aio_cu_pptr[col] = (IMCU **)palloc(sizeof(IMCU *) * MAX_CU_WRITE_REQSIZ);
            m_aio_dispath_cudesc[col] =
                (AioDispatchCUDesc_t **)palloc(sizeof(AioDispatchCUDesc_t *) * MAX_CU_WRITE_REQSIZ);
            m_vfdList[col] = (File *)palloc(sizeof(File) * MAX_CU_WRITE_REQSIZ);
            errno_t rc = memset_s((char *)m_vfdList[col], sizeof(File) * MAX_CU_WRITE_REQSIZ, 0xFF,
                sizeof(File) * MAX_CU_WRITE_REQSIZ);
            securec_check(rc, "\0", "\0");
        }
    }
    ADIO_END();
}

void IMCSInsert::InitDeltaInfo()
{
    // Oid delta_oid = m_relation->rd_rel->reldeltarelid;
    // m_delta_relation = relation_open(delta_oid, RowExclusiveLock); lqytodo create and init delta
    // m_delta_desc = m_delta_relation->rd_att;
}

/*
 * @Description: init index key logical column id.
 * @See also:
 */
void IMCSInsert::InitIndexColId(int which_index)
{
    int attrs_num = td_tid->natts;
    FormData_pg_attribute *attrs = td_tid->attrs;
    AttrNumber *key_ptr = NULL;
    List *expr_list = NULL;

    key_ptr = m_resultRelInfo->ri_IndexRelationInfo[which_index]->ii_KeyAttrNumbers;
    m_idxKeyNum[which_index] = m_resultRelInfo->ri_IndexRelationInfo[which_index]->ii_NumIndexAttrs;
    m_idxKeyAttr[which_index] = (int *)palloc(sizeof(int) * m_idxKeyNum[which_index]);
    expr_list = m_resultRelInfo->ri_IndexRelationInfo[which_index]->ii_Expressions;

    if (expr_list == NIL) {
        for (int which_idxkey = 0; which_idxkey < m_idxKeyNum[which_index]; ++which_idxkey) {
            for (int attr_cnt = 0; attr_cnt < attrs_num; ++attr_cnt) {
                if (attrs[attr_cnt].attnum == key_ptr[which_idxkey]) {
                    m_idxKeyAttr[which_index][which_idxkey] = attr_cnt;
                    break;
                }
            }
        }
    } else {
        List *vars = pull_var_clause((Node *)expr_list, PVC_RECURSE_AGGREGATES, PVC_RECURSE_PLACEHOLDERS);
        ListCell *lc = list_head(vars);
        for (int which_idxkey = 0; which_idxkey < m_idxKeyNum[which_index]; ++which_idxkey) {
            int keycol = key_ptr[which_idxkey];
            if (keycol == 0) {
                Var *var = (Var *)lfirst(lc);
                keycol = (int)var->varattno;
                lc = lnext(lc);
            }

            for (int attr_cnt = 0; attr_cnt < attrs_num; ++attr_cnt) {
                if (attrs[attr_cnt].attnum == keycol) {
                    m_idxKeyAttr[which_index][which_idxkey] = attr_cnt;
                    break;
                }
            }
        }
        list_free(vars);
    }
}

/*
 * @Description: init index insert info.
 * @See also:
 */
void IMCSInsert::InitIndexInfo(void)
{
    m_idxInsert = NULL;
    m_idxInsertArgs = NULL;

    if (relation_has_indexes(m_resultRelInfo)) {
        /* allocate memory for basic member variables */
        m_idxInsert = (IMCSInsert **)palloc0(sizeof(IMCSInsert *) * m_resultRelInfo->ri_NumIndices);
        m_idxInsertArgs = (IMInsertArg *)palloc0(sizeof(IMInsertArg) * m_resultRelInfo->ri_NumIndices);
        m_idxKeyAttr = (int **)palloc(sizeof(int *) * m_resultRelInfo->ri_NumIndices);
        m_idxKeyNum = (int *)palloc(sizeof(int) * m_resultRelInfo->ri_NumIndices);
        m_fake_values = (Datum *)palloc(sizeof(Datum) * td_tid->natts);
        m_fake_isnull = (bool *)palloc(sizeof(bool) * td_tid->natts);
        m_estate = CreateExecutorState();
        m_econtext = GetPerTupleExprContext(m_estate);
        RelationPtr idx_rel_array = m_resultRelInfo->ri_IndexRelationDescs;
        m_idxRelation = (Relation *)palloc(sizeof(Relation) * m_resultRelInfo->ri_NumIndices);
        m_deltaIdxRelation = (Relation *)palloc0(sizeof(Relation) * m_resultRelInfo->ri_NumIndices);

        bool is_partition = RelationIsPartition(m_relation);

        /* Loop all index and get some information */
        for (int which_index = 0; which_index < m_resultRelInfo->ri_NumIndices; ++which_index) {
            Oid idx_oid;
            Relation raw_index_rel = idx_rel_array[which_index];

            /* PSort index oid is stored in index_rel->relcudescrelid.
             * We reuse relcudescrelid as real index talbe oid.
             */
            if (is_partition) {
                Oid partidx_oid = getPartitionIndexOid(RelationGetRelid(raw_index_rel), RelationGetRelid(m_relation));
                Partition part_index = partitionOpen(raw_index_rel, partidx_oid, RowExclusiveLock);

                if (raw_index_rel->rd_rel->relam == PSORT_AM_OID)
                    idx_oid = part_index->pd_part->relcudescrelid;
                else {
                    m_idxRelation[which_index] = partitionGetRelation(raw_index_rel, part_index);
                    idx_oid = InvalidOid;
                }
                partitionClose(raw_index_rel, part_index, NoLock);
            } else {
                if (raw_index_rel->rd_rel->relam == PSORT_AM_OID)
                    idx_oid = raw_index_rel->rd_rel->relcudescrelid;
                else
                    idx_oid = RelationGetRelid(raw_index_rel);
            }

            /* open this index relation */
            if (idx_oid != InvalidOid) {
                m_idxRelation[which_index] = relation_open(idx_oid, RowExclusiveLock);
#ifndef ENABLE_MULTI_NODES
            }
#endif
            IMCSInsert::InitIndexColId(which_index);

            if (raw_index_rel->rd_rel->relam == PSORT_AM_OID) {
                /* Initilize index relation' batch buffer */
                IMCSInsert::InitIndexInsertArg(m_relation, m_idxKeyAttr[which_index], m_idxKeyNum[which_index],
                    m_idxInsertArgs[which_index]);
                /* Initilize index inserter */
                // useless
                // m_idxInsert[which_index] = New(CurrentMemoryContext)
                //                             IMCSInsert(m_idxRelation[which_index], m_idxInsertArgs[which_index],
                //                             false, NULL, NULL);
            }
        }
    }
}

void IMCSInsert::EndBatchInsert()
{
    int attNo = td_tid->natts;

    ADIO_RUN()
    {
        CUListFlushAll(attNo);
    }
    ADIO_ELSE()
    {
        // for (int i = 0; i < attNo && m_cuStorage[i] != NULL; ++i)
        //     m_cuStorage[i]->flush_data_file();
    }
    ADIO_END();
}

void IMCSInsert::BatchInsertCommon(IMCSBulkloadRows *batch_row_ptr, IMCU **old_cus, IMCUDesc **old_cus_descs,
    bool multithread, IMCSDesc *imcs_desc)
{
    if (unlikely(batch_row_ptr == NULL || batch_row_ptr->m_rows_curnum == 0))
        return;

    int col = 0;
    bool use_exist_imcu_cache_desc = false;

    CHECK_FOR_INTERRUPTS();
    /* step 1: form IMCU and IMCUDesc; */
    for (col = 0; col < batch_row_ptr->m_attr_num; ++col) {
        if (!td_tid->attrs[col].attisdropped) {
            m_cuCmprsOptions[col].m_sampling_finished = true;
            m_cuPPtr[col] = FormCU(col, batch_row_ptr, m_cuDescPPtr[col], td_tid, old_cus_descs == nullptr);
        }
    }
    if (m_isUpdate)
        pgstat_count_cu_update(m_relation, batch_row_ptr->m_rows_curnum);
    else
        pgstat_count_cu_insert(m_relation, batch_row_ptr->m_rows_curnum);

    /*
     * step 2: a) Allocate CUID and CUPointer
     * 		   b) Write IMCU and IMCUDesc
     */

    uint32 max_cu_id = 0;
    IMCU_CACHE->AcquireImcsDescLock(RelationGetRelid(m_relation), LW_EXCLUSIVE);
    // init cu_pointer and cudesc
    for (col = 0; col < batch_row_ptr->m_attr_num; ++col) {
        if (td_tid->attrs[col].attisdropped)
            continue;
        /* step 2: Allocate space, update m_cuDescPPtr[col]->cu_pointer */
        IMCUDesc *cu_desc = m_cuDescPPtr[col];

        int imcs_key = td_tid->attrs[col].attnum;
        if (imcs_key == VirtualCtidColID) {
            imcs_key = 0;
        }
        cu_desc->cu_id = m_cur_cuid;

        if (likely(cu_desc->cu_size > 0)) {
            cu_desc->cu_pointer = (uint64)0;
        }

        use_exist_imcu_cache_desc = false;

        if (!use_exist_imcu_cache_desc) {
            bool has_found = false;
            DataSlotTag data_slot_tag = IMCU_CACHE->InitCUSlotTag((RelFileNodeOld *)&m_relation->rd_node, imcs_key,
                cu_desc->cu_id, cu_desc->cu_pointer);
            if (!cu_desc->IsNullCU()) {
                m_slot_id = IMCU_CACHE->ReserveImcuBlock(&data_slot_tag, cu_desc, has_found, m_cuPPtr[col]);
            }

            int attlen = td_tid->attrs[col].attlen;
            max_cu_id = IMCU_CACHE->SaveImcuDesc(RelationGetRelid(m_relation), cu_desc, imcs_key, attlen);
        }

        if (m_cuPPtr[col]) {
            m_cuPPtr[col] = NULL;
        }
    }

    // increase row group id
    IMCU_CACHE->IncreaseImcsRowGroupNum(RelationGetRelid(m_relation), max_cu_id);
    m_cur_cuid++;

    IMCU_CACHE->ReleaseImcsDescLock(RelationGetRelid(m_relation));
}

/*
 * @Description: all columns flush cus which cached
 * @Param[IN] attno: relation column count
 * @See also:
 */
void IMCSInsert::CUListFlushAll(int attno)
{
#ifndef ENABLE_LITE_MODE
    /* flush cu */
    for (int col = 0; col < attno; ++col) {
        int count = m_aio_dispath_idx[col];
        AioDispatchCUDesc_t **d_list = m_aio_dispath_cudesc[col];
        /* submint io */
        if (count > 0) {
            FileAsyncCUWrite(d_list, count);
        }
    }

    /* check flush state */
    for (int col = 0; col < attno; ++col) {
        int count = m_aio_dispath_idx[col];
        if (count > 0) {
            CUListWriteCompeleteIO(col, count);
            count = 0;
        }
        m_aio_dispath_idx[col] = count;
    }
#endif
}

/*
 * @Description: async write one cu
 * @Param[IN] attno: relation column count
 * @Param[IN] col:column idx
 * @See also:
 */
void IMCSInsert::CUWrite(int attno, int col)
{
#ifndef ENABLE_LITE_MODE
    IMCU *cu = m_cuPPtr[col];
    IMCUDesc *cu_desc = m_cuDescPPtr[col];
    IMCUStorage *cu_storage = m_cuStorage[col];
    AioDispatchCUDesc_t **d_list = m_aio_dispath_cudesc[col];
    int count = m_aio_dispath_idx[col];
    AioDispatchCUDesc_t *aio_descp = NULL;

    Assert(cu_desc->cu_size > 0);

    /* if the cu store in two files or more, need flush the cus cached and sync write the cu */
    if (!cu_storage->IsCUStoreInOneFile(cu_desc->cu_pointer, cu_desc->cu_size)) {
        /* submint io */
        if (count > 0) {
            FileAsyncCUWrite(d_list, count);
            CUListWriteCompeleteIO(col, count);
            count = 0;
        }
        m_aio_dispath_idx[col] = count;

        cu_storage->SaveCU(cu->m_compressedBuf, cu_desc->cu_pointer, cu_desc->cu_size, true);
        CStoreCUReplication(m_relation, cu_storage->m_cnode.m_attid, cu->m_compressedBuf, cu_desc->cu_size,
            cu_desc->cu_pointer);
        cu->FreeMem<false>();
        ereport(LOG,
            (errmodule(MOD_ADIO), errmsg("CUListWrite: sync write cloumn(%d), cuid(%u), offset(%lu), size(%d)  ", col,
            cu_desc->cu_id, cu_desc->cu_pointer, cu_desc->cu_size)));
        return;
    }

    /* when cache cu exceed upper limit, write all */
    bool need_flush_cache = false;
    IMCU_CACHE->LockPrivateCache();
    if (hstore_adio_write_cache_size + cu_desc->cu_size >=
        u_sess->attr.attr_storage.cstore_backwrite_max_threshold * 1024LL) {
        need_flush_cache = true;
    }
    IMCU_CACHE->UnLockPrivateCache();
    if (need_flush_cache) {
        ereport(LOG, (errmodule(MOD_ADIO), errmsg("CUListWrite: exceed total cache, relid(%u), size(%ld)  ",
            RelationGetRelid(m_relation), hstore_adio_write_cache_size)));
        CUListFlushAll(attno);
    }

    /* must get count again becaue CUListFlushAll change  m_aio_dispath_idx[col] */
    count = m_aio_dispath_idx[col];

    /* when col cache cu exceed upper limit, write cache cu of this col */
    if (m_aio_cache_write_threshold[col] + cu_desc->cu_size >= cstore_backwrite_quantity * 1024LL) {
        ereport(DEBUG1,
            (errmodule(MOD_ADIO), errmsg("CUListWrite: exceed write threshold, column(%d), count(%d), size(%d)  ", col,
            count, m_aio_cache_write_threshold[col])));
        /* submint io */
        if (count > 0) {
            FileAsyncCUWrite(d_list, count);
            CUListWriteCompeleteIO(col, count);
            count = 0;
        }
    }
    m_aio_cache_write_threshold[col] += cu_desc->cu_size;

    aio_descp = (AioDispatchCUDesc_t *)adio_share_alloc(sizeof(AioDispatchCUDesc_t));

    m_aio_cu_pptr[col][count] = cu;

    /* iocb filled in later */
    aio_descp->aiocb.data = 0;
    aio_descp->aiocb.aio_fildes = 0;
    aio_descp->aiocb.aio_lio_opcode = 0;
    aio_descp->aiocb.u.c.buf = 0;
    aio_descp->aiocb.u.c.nbytes = 0;
    aio_descp->aiocb.u.c.offset = 0;

    aio_descp->cuDesc.buf = cu->m_compressedBuf;
    aio_descp->cuDesc.offset = cu_storage->GetCUOffsetInFile(cu_desc->cu_pointer);
    aio_descp->cuDesc.size = cu_desc->cu_size;
    aio_descp->cuDesc.fd = cu_storage->GetCUFileFd(cu_desc->cu_pointer);
    m_vfdList[col][count] = aio_descp->cuDesc.fd;
    aio_descp->cuDesc.io_error = &cu->m_adio_error;
    aio_descp->cuDesc.slotId = cu_desc->cu_id;
    aio_descp->cuDesc.cu_pointer = cu_desc->cu_pointer;
    aio_descp->cuDesc.io_finish = false;
    aio_descp->cuDesc.reqType = CUListWriteType;
    aio_descp->aiocb.aio_reqprio = CompltrPriority(aio_descp->cuDesc.reqType);

    d_list[count] = aio_descp;
    io_prep_pwrite((struct iocb *)d_list[count], aio_descp->cuDesc.fd, aio_descp->cuDesc.buf, aio_descp->cuDesc.size,
        aio_descp->cuDesc.offset);
    count++;

    /* record size */
    IMCU_CACHE->LockPrivateCache();
    hstore_adio_write_cache_size += cu_desc->cu_size;
    IMCU_CACHE->UnLockPrivateCache();
    ereport(DEBUG1,
        (errmodule(MOD_ADIO), errmsg("CUListWrite: increase cache size, column(%d), cu_size(%d),total cache size(%ld)",
        col, cu_desc->cu_size, hstore_adio_write_cache_size)));

    /* submint io */
    if (count >= MAX_CU_WRITE_REQSIZ) {
        ereport(DEBUG1, (errmodule(MOD_ADIO), errmsg("CUListWrite: exceed queue size, cloumn(%d), count(%d), size(%d) ",
            col, count, m_aio_cache_write_threshold[col])));
        FileAsyncCUWrite(d_list, count);
        CUListWriteCompeleteIO(col, count);
        count = 0;
    }
    m_aio_dispath_idx[col] = count;
    ereport(DEBUG1, (errmodule(MOD_ADIO), errmsg("CUListWrite: add cloumn(%d), cuid(%u), offset(%lu), size(%d) ", col,
        cu_desc->cu_id, cu_desc->cu_pointer, cu_desc->cu_size)));

    return;
#endif
}

/*
 * @Description:  write all column cu
 * @See also:
 */
void IMCSInsert::CUListWrite()
{
    int attno = m_relation->rd_rel->relnatts;
    int col = 0;

    PG_ENSURE_ERROR_CLEANUP(CUListWriteAbort, (Datum)this);
    {
        for (col = 0; col < attno; ++col) {
            if (!td_tid->attrs[col].attisdropped) {
                if (m_cuDescPPtr[col]->cu_size > 0) {
                    CUWrite(attno, col);
                } else {
                    /* same cu and null cu need free memory early */
                    DELETE_EX(m_cuPPtr[col]);
                }
            }
        }
    }
    PG_END_ENSURE_ERROR_CLEANUP(CUListWriteAbort, (Datum)this);
    return;
}


/*
 * @Description: wait the async write cu finish and check write state, then send to stanby and free resource
 * @Param[IN] col: column id
 * @Param[IN] count: column cache cu count
 * @See also:
 */
void IMCSInsert::CUListWriteCompeleteIO(int col, int count)
{
#ifndef ENABLE_LITE_MODE
    int idx = 0;
    AioDispatchCUDesc_t **d_list = m_aio_dispath_cudesc[col];
    IMCUStorage *cu_storage = m_cuStorage[col];

    for (idx = 0; idx < count; ++idx) {
        IMCU *cu = m_aio_cu_pptr[col][idx];
        AioCUDesc_t *cu_desc = &(d_list[idx]->cuDesc);

        while (!cu_desc->io_finish) {
            /* see jack email, low efficient, think more */
            pg_usleep(1);
        }

        if (cu->m_adio_error) {
            ereport(ERROR, (errcode_for_file_access(),
                errmsg("write cu failed, colid(%d) cuid(%u), offset(%lu), size(%d) : %m", col, (uint32)cu_desc->slotId,
                cu_desc->cu_pointer, cu_desc->size),
                errhint("Check free disk space.")));
        }
    }

    for (idx = 0; idx < count; ++idx) {
        IMCU *cu = m_aio_cu_pptr[col][idx];
        AioCUDesc_t *cu_desc = &(d_list[idx]->cuDesc);

        if (cu_desc->size > 0) {
            CStoreCUReplication(m_relation, cu_storage->m_cnode.m_attid, cu->m_compressedBuf, cu_desc->size,
                cu_desc->cu_pointer);
            cu->FreeMem<false>();

            IMCU_CACHE->LockPrivateCache();
            hstore_adio_write_cache_size -= cu_desc->size;
            Assert(hstore_adio_write_cache_size >= 0);
            IMCU_CACHE->UnLockPrivateCache();
            ereport(DEBUG1, (errmodule(MOD_ADIO), errmsg("CUListWriteCompeleteIO: decrease cache size, column(%d),"
                "idx(%d), total cache size(%ld)",
                col, idx, hstore_adio_write_cache_size)));
            m_aio_cache_write_threshold[col] -= cu_desc->size;
            Assert(m_aio_cache_write_threshold[col] >= 0);
        }
        DELETE_EX(cu);
        adio_share_free((void *)d_list[idx]);
        d_list[idx] = NULL;
    }

    FileAsyncCUClose(m_vfdList[col], count);
#endif
}

/* Write IMCU data and IMCUDesc */
void IMCSInsert::SaveAll(int options, _in_ const char *del_bitmap)
{
    int attno = m_relation->rd_rel->relnatts, col = 0;
    uint64 total_size = 0;
    int first_col_idx = 0;

    /*
     * IO Collector and IO Scheduler
     * for alloc_space, attno numbers of IO requests are to send to OS at most.
     * IO is controlled before acquring lock
     */
    if (ENABLE_WORKLOAD_CONTROL)
        IOSchedulerAndUpdate(IO_TYPE_WRITE, attno, IO_TYPE_COLUMN);

    STORAGE_SPACE_OPERATION(m_relation, 0);

    /* Ensure rd_smgr is open (could have been closed by relcache flush!) */
    RelationOpenSmgr(m_relation);

    /* step 1: Lock relation for extension */
    LockRelationForExtension(m_relation, ExclusiveLock);

    for (col = 0; col < attno; ++col) {
        if (td_tid->attrs[col].attisdropped)
            continue;
        /* step 2: Allocate space, update m_cuDescPPtr[col]->cu_pointer */
        IMCUDesc *cu_desc = m_cuDescPPtr[col];
        // IMCUStorage* cu_storage = m_cuStorage[col];
        first_col_idx = col;

        cu_desc->cu_id = ++m_cur_cuid;
        if (likely(cu_desc->cu_size > 0)) {
            // cu_desc->cu_pointer = cu_storage->alloc_space(cu_desc->cu_size);
            cu_desc->cu_pointer = (uint64)palloc0(cu_desc->cu_size);

            total_size += cu_desc->cu_size;
        }

        /* step 3: Save IMCUDesc */
        // IMCStore::save_cu_desc(m_relation, cu_desc, col, options);
    }

    /* storage space processing before copying column data. */
    perm_space_increase(m_relation->rd_rel->relowner, total_size,
        RelationUsesSpaceType(m_relation->rd_rel->relpersistence));

    /* step 4: unlock extension locker */
    UnlockRelationForExtension(m_relation, ExclusiveLock);

    /* Step 5: Write IMCU into storage */
    ADIO_RUN()
    {
        CUListWrite();
    }
    ADIO_ELSE()
    {
        for (col = 0; col < attno; ++col) {
            if (td_tid->attrs[col].attisdropped) {
                if (m_cuPPtr[col])
                    DELETE_EX(m_cuPPtr[col]);
                continue;
            }
            IMCU *cu = m_cuPPtr[col];
            IMCUDesc *cu_desc = m_cuDescPPtr[col];
            IMCUStorage *cu_storage = m_cuStorage[col];

            if (cu_desc->cu_size > 0) {
                /* IO collector and IO scheduler */
                if (ENABLE_WORKLOAD_CONTROL)
                    IOSchedulerAndUpdate(IO_TYPE_WRITE, 1, IO_TYPE_COLUMN);

                cu_storage->SaveCU(cu->m_compressedBuf, cu_desc->cu_pointer, cu_desc->cu_size, false);
                const int CU_ALIGN_SIZE = cu_storage->Is2ByteAlign() ? ALIGNOF_TIMESERIES_CUSIZE : ALIGNOF_CUSIZE;
                if (u_sess->attr.attr_storage.HaModuleDebug)
                    ereport(LOG, (errmsg("HA-save_all: rnode %u/%u/%u, col %d, blockno %lu, cu_unit_count %d",
                        m_relation->rd_node.spcNode, m_relation->rd_node.dbNode, m_relation->rd_node.relNode,
                        cu_storage->m_cnode.m_attid, cu_desc->cu_pointer / CU_ALIGN_SIZE,
                        cu_desc->cu_size / CU_ALIGN_SIZE)));

                CStoreCUReplication(m_relation, cu_storage->m_cnode.m_attid, cu->m_compressedBuf, cu_desc->cu_size,
                    cu_desc->cu_pointer);

                cu->FreeMem<false>();
            }

            /* free IMCU object immediately */
            if (m_cuPPtr[col])
                DELETE_EX(m_cuPPtr[col]);
        }
    }
    ADIO_END();

    /* step 6: Insert IMCUDesc of virtual column */
    // IMCStore::save_vc_cu_desc(m_relation->rd_rel->relcudescrelid, m_cuDescPPtr[first_col_idx]->cu_id,
    //                      m_cuDescPPtr[first_col_idx]->row_count, m_cuDescPPtr[first_col_idx]->magic, options,
    //                      del_bitmap);

    /* Workaround for those SQL (insert) unsupported by optimizer */
    if (unlikely(!IS_PGXC_DATANODE)) {
        ereport(ERROR,
            (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), (errmsg("This query is not supported by optimizer in CStore."))));
    }
}

/*
 * @Description: form IMCU data
 * @IN batch_row_ptr: batchrows
 * @IN col: which column to handle
 * @IN/OUT cuDescPtr: IMCU Descriptor object
 * @Return: IMCU object
 * @See also:
 */
IMCU *IMCSInsert::FormCU(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, TupleDesc tupdesc_with_ctid,
    bool compress = true)
{
    FormData_pg_attribute *attrs = tupdesc_with_ctid->attrs;
    int attlen = attrs[col].attlen;
    IMCU *cu_ptr = NULL;

    ADIO_RUN()
    {
        /* cu_ptr need keep untill async write finish */
        cu_ptr = New(m_aio_memcnxt)IMCU(attlen, attrs[col].atttypmod, attrs[col].atttypid);
    }
    ADIO_ELSE()
    {
        cu_ptr = New(CurrentMemoryContext)IMCU(attlen, attrs[col].atttypmod, attrs[col].atttypid);
    }
    ADIO_END();

    cuDescPtr->Reset();

    int fun_idx = batch_row_ptr->m_vectors[col].m_values_nulls.m_has_null ? FORMCU_IDX_HAVE_NULL : FORMCU_IDX_NONE_NULL;
    (this->*(m_formCUFuncArray[col].col_form_cu[fun_idx]))(col, batch_row_ptr, cuDescPtr, cu_ptr);

    // We should not compress in two case.
    // case1) IsNullCU
    // case2) Min is the same to max in CU. In this case, we don't
    // 		  need IMCUStorage
    cuDescPtr->magic = GetCurrentTransactionIdIfAny();
    if (!(cuDescPtr->IsNullCU())) {
        // a little tricky to reduce the recomputation of min/max value.
        // some data type is equal to int8/int16/int32/int32. for them it
        // is not necessary to recompute the min/max value.
        m_cuTempInfo.m_valid_minmax = !NeedToRecomputeMinMax(attrs[col].atttypid);
        if (m_cuTempInfo.m_valid_minmax) {
            m_cuTempInfo.m_min_value = ConvertToInt64Data(cuDescPtr->cu_min, attlen);
            m_cuTempInfo.m_max_value = ConvertToInt64Data(cuDescPtr->cu_max, attlen);
        }
        m_cuTempInfo.m_options = (m_cuCmprsOptions + col);
        cu_ptr->m_tmpinfo = &m_cuTempInfo;

        /* Magic number is for checking IMCU data */
        cu_ptr->SetMagic(cuDescPtr->magic);
        if (compress) {
            cu_ptr->Compress(batch_row_ptr->m_rows_curnum, m_compress_modes, ALIGNOF_CUSIZE);
            cuDescPtr->cu_size = cu_ptr->GetCUSize();
        }
    }
    cuDescPtr->row_count = batch_row_ptr->m_rows_curnum;
    cuDescPtr->flag = IMCUDescInMemory;

    return cu_ptr;
}

/*
 * @Description: encode numeric values
 * @IN batch_row_ptr: batch values about numeric
 * @IN col: which column
 * @IN/OUT cuDescPtr: IMCU Description
 * @IN/OUT cu_ptr: IMCU object
 * @IN has_null: null flag
 * @Return: whether encode numeric values successfully
 * @See also:
 */
bool IMCSInsert::TryEncodeNumeric(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, IMCU *cu_ptr,
    bool has_null)
{
    char *ptr = NULL;
    int rows = batch_row_ptr->m_rows_curnum;
    int not_exact_size = 0;
    int exact_size = 0;
    int rc = 0;

    if (COMPRESS_NO == heaprel_get_compression_from_modes(m_compress_modes)) {
        /* nothing to do if compression level is NO. */
        return false;
    }

    /* description: future plan-memory opt. */
    numerics_statistics_out phase1_out;
    rc = memset_s(&phase1_out, sizeof(numerics_statistics_out), 0, sizeof(numerics_statistics_out));
    securec_check(rc, "\0", "\0");
    phase1_out.success = (bool *)palloc(sizeof(bool) * rows);
    phase1_out.ascale_codes = (char *)palloc(sizeof(char) * rows);
    /* In order to avoid over-boundary read in the function readData, more 8 byte memory is allocated. */
    phase1_out.int64_values = (int64 *)palloc(sizeof(int64) * rows + 8);
    phase1_out.int32_values = (int32 *)palloc(sizeof(int32) * rows + 8);

    numerics_cmprs_out phase2_out;
    rc = memset_s(&phase2_out, sizeof(numerics_cmprs_out), 0, sizeof(numerics_cmprs_out));
    securec_check(rc, "\0", "\0");

    /* set compression filter which is from bulkload inserter */
    phase2_out.filter = m_cuCmprsOptions + col;

    BatchNumeric batch = { batch_row_ptr->m_vectors[col].m_values_nulls.m_vals_points,
                           batch_row_ptr->m_vectors[col].m_values_nulls.m_null_bitmap,
                           batch_row_ptr->m_rows_curnum,
                           has_null,
                           CuAppendNullValue,
                           (void *)cu_ptr };
    if (NumericCompressBatchValues(&batch, &phase1_out, &phase2_out, &not_exact_size, ALIGNOF_CUSIZE)) {
        /* expand the memory if needed */
        Assert(cu_ptr->m_srcDataSize == 0);
        if (unlikely(cu_ptr->m_srcData + not_exact_size > cu_ptr->m_srcBuf + cu_ptr->m_srcBufSize)) {
            Assert(not_exact_size >= 0);
            cu_ptr->ReallocMem((Size)cu_ptr->m_srcBufSize + (uint32)not_exact_size);
        }

        ptr = NumericCopyCompressedBatchValues(cu_ptr->m_srcData, &phase1_out, &phase2_out);

        if (!m_cuCmprsOptions[col].m_sampling_finished) {
            /* get adopted compression methods from the first IMCU sample */
            m_cuCmprsOptions[col].set_numeric_flags(*(uint16 *)cu_ptr->m_srcData);
        }

        /// correct the import information with CU object.
        exact_size = (ptr - cu_ptr->m_srcData);

        /// Important:
        /// now for numeric compression, *m_srcDataSize* is the size of compressed
        /// integer values, but not the total size of raw numeric values which is
        /// remembered in *phase1_out.original_size*. so consider this during the
        /// decompression.
        cu_ptr->m_srcDataSize = exact_size;

        (void)IMCStore::SetCudescModeForMinMaxVal(false, false, has_null, 0, -1, cuDescPtr);

        /* release all the memory unused. */
        NumericCompressReleaseResource(&phase1_out, &phase2_out);

        // use d-scale int64 for numeric CU store.
        // and add CU_DSCALE_NUMERIC flags to cuPtr->m_infoMode
        cu_ptr->m_infoMode |= (CU_IntLikeCompressed | CU_DSCALE_NUMERIC);

        return true;
    }

    /* release all the memory unused. */
    NumericCompressReleaseResource(&phase1_out, &phase2_out);
    return false;
}

/*
 * this function change the simple string values to uint64 values, it requires first char from '1' to '9',
 * and others '0'~'9', and for fix length char type, ' ' in the behind is welcome.
 * '+' '_' ',' are not deal with this time.
 * strtoul() and snprintf() are low effcient for performance, but now,is ok.
 * this function is too long and use template for performance, so i did not split it into short function
 */
template <bool bpchar_type, bool has_null, bool tem_has_min_max_func>
bool IMCSInsert::FormNumberStringCU(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, IMCU *cu_ptr)
{
    int attlen = this->td_tid->attrs[col].attlen;
    int rows = batch_row_ptr->m_rows_curnum;
    imcs_bulkload_datums *batch_values = &(batch_row_ptr->m_vectors[col].m_values_nulls);
    uint64 data = 0;
    char *data_buf = (char *)palloc(sizeof(char) * (MAX_LEN_CHAR_TO_BIGINT_BUF) * (rows));
    uint32 data_count = 0;
    uint32 idx = 0;
    uint32 next_data_pos = 0;
    uint32 max_str_len = 0;
    uint32 bpchar_data_len = 0;
    /* the total length excludes the var-header size of all the values */
    uint32 total_len = 0;
    ImcsFuncSetMinMax *min_max_func = this->m_setMinMaxFuncs + col;
    bool first_flag = true;
    bool has_min_max_func = false;

    for (int i = 0; i < rows; ++i) {
        if (has_null && batch_values->is_null(i)) {
            cu_ptr->AppendNullValue(i);
            continue;
        }

        Datum v = batch_values->get_datum(i);
        char *p = DatumGetPointer(v);
        Assert(p != NULL);
        uint32 len = VARSIZE_ANY_EXHDR(p);
        /* this means that it's an empty string.
         * because we cannot map an empty string to any integer
         * so don't handle this case and return false.
         */
        if (unlikely(len == 0)) {
            pfree(data_buf);
            return false;
        }
        Assert(len > 0);

        total_len += len;
        /* fix length char type, need know origin data len */
        if (bpchar_type) {
            bpchar_data_len = len;
        }
        /* dymic length char type, check length first */
        if (!bpchar_type && len > MAX_LEN_CHAR_TO_BIGINT) {
            pfree(data_buf);
            return false;
        }
        if (max_str_len < len + VARHDRSZ) {
            max_str_len = len + VARHDRSZ;
        }
        char *ptr = VARDATA_ANY(p);
        /* first byte must be 1~9 */
        if (*ptr > '9' || *ptr < '1') {
            pfree(data_buf);
            return false;
        }

        /* this index used to write cannot be overflow. */
        Assert(idx < MAX_LEN_CHAR_TO_BIGINT_BUF * (uint32)rows);

        /* data_buf store valid data num */
        data_buf[idx] = *ptr;
        --len;
        ++ptr;
        ++idx;
        while (len > 0) {
            /* next byte must be 0~9 or ' ' */
            if (*ptr > '9' || *ptr < '0') {
                if (bpchar_type) {
                    /* fix length char, need check ' ' after num */
                    break;
                } else {
                    /* dymic length char can return not ok */
                    pfree(data_buf);
                    return false;
                }
            }

            /* fix length char, need check length and ' ' which all the buf left.
             * we will write extra one byte next step, so we have to check
             * the current length of digits. because not all 20 digits can be
             * converted to uint64 value, we have to finish if this case happens.
             */
            if (bpchar_type && bpchar_data_len - len >= MAX_LEN_CHAR_TO_BIGINT) {
                pfree(data_buf);
                return false;
            }

            data_buf[idx] = *ptr;
            --len;
            ++ptr;
            ++idx;
        }

        /* this index used to write cannot be overflow. */
        Assert(idx < MAX_LEN_CHAR_TO_BIGINT_BUF * (uint32)rows);
        if (bpchar_type) {
            /* each integer digits cannot be greater than 19 */
            Assert(bpchar_data_len - len <= MAX_LEN_CHAR_TO_BIGINT);
            MEMCTL_LOG(DEBUG2, "Assert each integer digits cannot be greater than 19 when bpchar_type is true");
        }

        /* set complete flag for function strtoul need it */
        data_buf[idx] = '\0';
        if (bpchar_type) {
            while (len > 0) {
                if (*ptr != ' ') {
                    pfree(data_buf);
                    return false;
                }
                --len;
                ++ptr;
            }
        }

        /* make right idx for store next data value */
        ++data_count;
        next_data_pos += MAX_LEN_CHAR_TO_BIGINT_BUF;
        idx = next_data_pos;

        if (tem_has_min_max_func) {
            (*(min_max_func))(v, cuDescPtr, &first_flag);
            has_min_max_func = true;
        }
        first_flag = false;
    }

    /* for deform, store count */
    cu_ptr->AppendValue(data_count, sizeof(uint64));

    /* do convert, must sucess */
    char *tmp_data_buf = data_buf;
    for (uint32 i = 0; i < data_count; ++i) {
        int ret = str_to_uint64(tmp_data_buf, &data);
        if (ret != 0) {
            pfree(data_buf);
            return false;
        }
        /* append each integer value */
        cu_ptr->AppendValue(data, sizeof(uint64));
        tmp_data_buf += MAX_LEN_CHAR_TO_BIGINT_BUF;
    }

    /* for deform, store src data size but excludes
     * their var-header size.
     */
    cu_ptr->AppendValue(total_len, sizeof(uint64));

    (void)IMCStore::SetCudescModeForMinMaxVal(first_flag, has_min_max_func, has_null, max_str_len, attlen, cuDescPtr);

    pfree(data_buf);
    cu_ptr->m_infoMode |= CU_IntLikeCompressed;
    return true;
}

template <bool has_null>
bool IMCSInsert::TryFormNumberStringCU(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, IMCU *cu_ptr,
    uint32 atttypid)
{
    bool ret = false;
    ImcsFuncSetMinMax func = *(this->m_setMinMaxFuncs + col);

    if (COMPRESS_NO == heaprel_get_compression_from_modes(this->m_compress_modes)) {
        return ret;
    }

    if (atttypid == BPCHAROID) {
        /* for type define bpchar, we do not know the length, so we don't do the change */
        if (this->td_tid->attrs[col].atttypmod == -1) {
            return ret;
        }
        if (func == NULL) {
            ret = this->FormNumberStringCU<true, has_null, false>(col, batch_row_ptr, cuDescPtr, cu_ptr);
        } else {
            ret = this->FormNumberStringCU<true, has_null, true>(col, batch_row_ptr, cuDescPtr, cu_ptr);
        }
    } else {
        if (func == NULL) {
            ret = this->FormNumberStringCU<false, has_null, false>(col, batch_row_ptr, cuDescPtr, cu_ptr);
        } else {
            ret = this->FormNumberStringCU<false, has_null, true>(col, batch_row_ptr, cuDescPtr, cu_ptr);
        }
    }

    return ret;
}

/*
 * @Description: init IMCU memory for copying
 * @IN batch_row_ptr: batchrows
 * @IN col: which column to handle
 * @IN/OUT cu_ptr: IMCU object
 * @IN has_null: has-null flag
 * @Return: total data size
 * @See also:
 */
Size IMCSInsert::FormCUTInitMem(IMCU *cu_ptr, IMCSBulkloadRows *batch_row_ptr, int col, bool has_null)
{
    imcs_bulkload_vector *vector = batch_row_ptr->m_vectors + col;

    /* compute the total size */
    Size dt_size = vector->data_size();
    Size initial_size = cu_ptr->GetCUHeaderSize() + sizeof(Datum) + dt_size;
    initial_size = MAXALIGN(initial_size);
    cu_ptr->InitMem((uint32)initial_size, batch_row_ptr->m_rows_curnum, has_null);
    return dt_size;
}

/*
 * @Description: copy and form IMCU data for common datatype
 * @IN batch_row_ptr: batchrows
 * @IN col: which column to handle
 * @IN/OUT cuDescPtr: IMCU Descriptor object
 * @IN/OUT cu_ptr: IMCU object
 * @IN dt_size: total data size
 * @IN has_null: has-null flag
 * @Return:
 * @See also:
 */
void IMCSInsert::FormCUTCopyMem(IMCU *cu_ptr, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, Size dt_size,
    int col, bool has_null)
{
    imcs_bulkload_vector *vector = batch_row_ptr->m_vectors + col;
    Form_pg_attribute attr = &this->td_tid->attrs[col];

    /* copy null-bitmap */
    if (has_null) {
        /* we have computed the total memory size including null bitmap */
        Size bpsize = bitmap_size(batch_row_ptr->m_rows_curnum);
        errno_t rc = memcpy_s(cu_ptr->m_nulls, bpsize, vector->m_values_nulls.m_null_bitmap, bpsize);
        securec_check(rc, "\0", "\0");
    }

    /* copy all the data */
    vector->data_copy(cu_ptr->m_srcData);
    cu_ptr->m_srcDataSize = dt_size;

    /* set min/max for IMCU Desc */
    bool has_min_max_func = !ImcsIsCompareDatumDummyFunc(vector->m_minmax.m_compare);
    int max_str_len = vector->m_minmax.m_varstr_maxlen;
    vector->m_minmax.m_finish_compare(vector->m_minmax.m_min_buf, vector->m_minmax.m_max_buf, cuDescPtr);

    /* Check whether IMCU is filled with the same value
     * Check whether IMCU is filled with all NULL value
     */
    (void)IMCStore::SetCudescModeForMinMaxVal(vector->m_values_nulls.m_all_null, has_min_max_func, has_null,
        max_str_len, attr->attlen, cuDescPtr);
}

/*
 * @Description: form IMCU data for common datatype
 * @IN batch_row_ptr: batchrows
 * @IN col: which column to handle
 * @IN/OUT cuDescPtr: IMCU Descriptor object
 * @IN/OUT cu_ptr: IMCU object
 * @Return:
 * @See also:
 */
template <bool has_null>
void IMCSInsert::FormCUT(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, IMCU *cu_ptr)
{
    Size dt_size = this->FormCUTInitMem(cu_ptr, batch_row_ptr, col, has_null);
    this->FormCUTCopyMem(cu_ptr, batch_row_ptr, cuDescPtr, dt_size, col, has_null);
}

/*
 * @Description: form IMCU data for numeric datatype
 * @IN batch_row_ptr: batchrows
 * @IN col: which column to handle
 * @IN/OUT cuDescPtr: IMCU Descriptor object
 * @IN/OUT cu_ptr: IMCU object
 * @Return:
 * @See also:
 */
template <bool has_null>
void IMCSInsert::FormCUTNumeric(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, IMCU *cu_ptr)
{
    Size dt_size = this->FormCUTInitMem(cu_ptr, batch_row_ptr, col, has_null);

    if (this->TryEncodeNumeric(col, batch_row_ptr, cuDescPtr, cu_ptr, has_null)) {
        return;
    }

    /* clean-up work here.
     * ignore the nulls memory even though it will be set later again.
     */
    cu_ptr->m_srcDataSize = 0;

    this->FormCUTCopyMem(cu_ptr, batch_row_ptr, cuDescPtr, dt_size, col, has_null);

    /* change m_formCUFuncArray[col] to FormCUTCommon()
     * if the first time of encoding-numeric fails
     */
    if (!this->m_cuCmprsOptions[col].m_sampling_finished) {
        this->m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_NONE_NULL] = &IMCSInsert::FormCUT<false>;
        this->m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_HAVE_NULL] = &IMCSInsert::FormCUT<true>;
    }
}

/*
 * @Description: form IMCU data for number string
 * @IN batch_row_ptr: batchrows
 * @IN col: which column to handle
 * @IN/OUT cuDescPtr: IMCU Descriptor object
 * @IN/OUT cu_ptr: IMCU object
 * @IN has_null: has-null flag
 * @Return:
 * @See also:
 */
template <bool has_null>
void IMCSInsert::FormCUTNumString(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cuDescPtr, IMCU *cu_ptr)
{
    Form_pg_attribute attr = &this->td_tid->attrs[col];

    Size data_size = this->FormCUTInitMem(cu_ptr, batch_row_ptr, col, has_null);

    if (this->TryFormNumberStringCU<has_null>(col, batch_row_ptr, cuDescPtr, cu_ptr, attr->atttypid)) {
        return;
    }

    /* clean-up work here. Ignore the nulls memory even though it will be set later again. */
    cu_ptr->m_srcDataSize = 0;

    this->FormCUTCopyMem(cu_ptr, batch_row_ptr, cuDescPtr, data_size, col, has_null);
    /* change m_formCUFuncArray[col] to FormCUTCommon()	if the first time of encoding-number-string fails */
    if (!this->m_cuCmprsOptions[col].m_sampling_finished) {
        this->m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_NONE_NULL] = &IMCSInsert::FormCUT<false>;
        this->m_formCUFuncArray[col].col_form_cu[FORMCU_IDX_HAVE_NULL] = &IMCSInsert::FormCUT<true>;
    }
}

/*
 * @Description: Interface for insertion with undecompressed cu_data. Modeled after
 * 			    BatchInsertCommon. Used only at_exec_cstore_merge_partition so far.
 * 			    Further application will require modification
 * @IN cu_data: IMCSBatchCUData that holds CUDescs, CUptrs, and Delbitmap
 * @IN options: insert option
 * @See also: BatchInsertCommon
 */
void IMCSInsert::CUInsert(_in_ IMCSBatchCUData *cu_data, _in_ int options)
{
    if (unlikely(cu_data == NULL))
        return;

    int attno = m_relation->rd_rel->relnatts;
    int col = 0;
    Assert(attno == cu_data->colNum);

    /* step 1: pass IMCUDesc and IMCU to IMCSInsert */
    for (col = 0; col < attno; ++col) {
        if (td_tid->attrs[col].attisdropped)
            continue;
        *m_cuDescPPtr[col] = *cu_data->CUDescData[col];
        m_cuPPtr[col] = cu_data->CUptrData[col];
    }

    /*
     * step 2: a) Allocate CUID and CUPointer
     * b) Write IMCU and IMCUDesc
     */
    SaveAll(options, cu_data->bitmap);
}


void IMCSInsert::SetEndFlag()
{
    m_insert_end_flag = true;
}

/*
 * @Description: init insert arguments for psort index
 * @OUT args:  insert argunets of batch buffer.
 * @IN heap_rel: heap relation
 * @IN key_map:  index keys map
 * @IN nkeys:    number of index keys
 * @See also:
 */
void IMCSInsert::InitIndexInsertArg(Relation heap_rel, const int *key_map, int nkeys, IMInsertArg &args)
{
    /* plus TID system attribute */
    int nkeys_plus_tid = nkeys + 1;
    errno_t rc;

    struct tupleDesc *index_tupdesc = CreateTemplateTupleDesc(nkeys_plus_tid, false);

    index_tupdesc->natts = nkeys_plus_tid;
    /* the following are not important to us, just init them */
    index_tupdesc->constr = NULL;
    index_tupdesc->initdefvals = NULL;
    index_tupdesc->tdhasoid = false;
    index_tupdesc->tdrefcount = 1;
    index_tupdesc->tdtypeid = InvalidOid;
    index_tupdesc->tdtypmod = -1;

    /* set attribute point exlcuding TID field */
    for (int i = 0; i < nkeys; ++i) {
        rc = memcpy_s(&index_tupdesc->attrs[i], ATTRIBUTE_FIXED_PART_SIZE, &heap_rel->rd_att->attrs[key_map[i]],
            ATTRIBUTE_FIXED_PART_SIZE);
        securec_check(rc, "\0", "\0");
    }


    /* set TID attribute */
    FormData_pg_attribute tid_attr;
    imcs_init_tid_attinfo(&tid_attr);
    rc = memcpy_s(&index_tupdesc->attrs[nkeys], ATTRIBUTE_FIXED_PART_SIZE, &tid_attr, ATTRIBUTE_FIXED_PART_SIZE);
    securec_check(rc, "\0", "\0");

    args.es_result_relations = NULL;

    /* psort index shouldn't have any index, so make it NULL */
    args.idx_batch_row = NULL;
    /* init temp batch buffer for psort index */
    args.tmp_batch_rows =
        New(CurrentMemoryContext)IMCSBulkloadRows(index_tupdesc, RelationGetMaxBatchRows(heap_rel), true);
    args.using_vectorbatch = false;

    /* release temp memory */
    pfree(index_tupdesc);
}

/*
 * @Description: init all kinds of batch buffers during bulk-load.
 * @OUT args: all kinds of batch buffers.
 * @IN using_vectorbatch: see IMInsertArg::using_vectorbatch comments.
 * @IN rel: relation to query
 * @IN result_rel_info: resulting relation info
 * @See also:
 */
void IMCSInsert::InitInsertArg(Relation rel, ResultRelInfo *result_rel_info, bool using_vectorbatch, IMInsertArg &args)
{
    TupleConstr *constr = rel->rd_att->constr;
    int max_values_count = RelationGetMaxBatchRows(rel);

    args.es_result_relations = result_rel_info;
    args.using_vectorbatch = using_vectorbatch;

    /* create temp batchrows if relation has PCK */
    if (tupledesc_have_pck(constr)) {
        args.tmp_batch_rows = New(CurrentMemoryContext)IMCSBulkloadRows(rel->rd_att, max_values_count, true);
    }

    /* create BatchRows for index, if relation has index. */
    if (relation_has_indexes(result_rel_info)) {
        /* find max number of index keys,
         * and we will reuse these batch buffers.
         */
        int max_key_num = get_max_num_of_index_keys(result_rel_info);
        args.idx_batch_row = imcs_bulkload_indexbatch_init(max_key_num, max_values_count);
    }
}

/*
 * @Description: destroy insert arguments
 * @IN args: insert arguments
 * @See also:
 */
void IMCSInsert::DeInitInsertArg(IMInsertArg &args)
{
    if (args.tmp_batch_rows) {
        DELETE_EX(args.tmp_batch_rows);
    }
    if (args.idx_batch_row) {
        imcs_bulkload_indexbatch_deinit(args.idx_batch_row);
    }
}
